package com.sxhuayuan.parking.service.impl;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.sxhuayuan.parking.exception.MyException;
import com.sxhuayuan.parking.plugin.parking.ParkingPlugin;
import com.sxhuayuan.parking.queue.msg.NotifyParkingLotMsg;
import com.sxhuayuan.parking.service.PluginService;
import com.sxhuayuan.parking.service.RedisService;
import com.sxhuayuan.parking.service.ThirdPartiesNotifyService;
import com.sxhuayuan.parking.utils.SettingUtils;
import com.sxhuayuan.parking.utils.SpringUtils;

//@DependsOn("redisServiceImpl")
//@Service("thirdPartiesNotifyServiceImpl")
@Deprecated
public class ThirdPartiesNotifyServiceImpl implements ThirdPartiesNotifyService, InitializingBean, DisposableBean {

	Logger log = LoggerFactory.getLogger(getClass());

	public static final String NOTIFY_EVENT_QUEUE_NAME = "thrdPartiesNotifyQueue";
	// 线程池缓冲队列大小
	private static final Integer QUEUE_SIZE = 10;
	// 线程池最大线程数
	private static final Integer MAX_POOL_SIZE = 1;
	// 线程池核心线程数
	private static final Integer CORE_POOL_SIZE = 1;

	@Resource(name = "redisServiceImpl")
	RedisService redisService;
	// 单线程，只负责从队列取出订单信息，启动通知线程
	private static ExecutorService queuePopTaskExecutor = null;
	// 上面线程的结果
	private static Future<?> queuePopTaskfuture;
	// 多线程池，负责执行http请求通知第三方
	private static ExecutorService notifyTaskExecutor = null;
	// 通知线程池的循环退出标志位
	private static boolean notifyTaskExitFlag = true;
	// 从队列中取值的间隔时间，如果消息产生速度大于消费速度，则需要减慢取出的速度，否则会出现消费不掉的消息又不断地取出放回redis循环此过程，造成网络和资源白白消耗。相反，这个值也不能太大，否则消费线程又会处于饿饭状态，所以如果生产速度大于消费速度时，此值应该趋近于某个平衡的值。
	private final AtomicInteger popDelayTime = new AtomicInteger();// ms

	@Override
	public boolean notifyOrderCompleted(String orderSn, String pluginId, String lotSn, String carPlate) {
		NotifyParkingLotMsg bean = new NotifyParkingLotMsg(pluginId, orderSn, lotSn, carPlate, "");
		String jsonString = JSON.toJSONString(bean);
		boolean isSuccess = redisService.leftpush(NOTIFY_EVENT_QUEUE_NAME, jsonString);
		if (!isSuccess) {
			log.error("push 回调事件失败: {}", jsonString);
			return false;
		}
		return true;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if (notifyTaskExecutor == null) {
			ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
				@Override
				public void uncaughtException(Thread t, Throwable e) {

				}
			}).build();
			LinkedBlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
			notifyTaskExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, 60, TimeUnit.SECONDS, workQueue, threadFactory, new RejectedExecutionHandler() {
				@Override
				public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
					// 每拒绝一次，队列pop间隔时间增加2ms，抵消取时的1ms，实际上等于失败一次增加1ms
					int delayMillions = popDelayTime.addAndGet(2);
					NotifyTask task = (NotifyTask) r;
					NotifyParkingLotMsg notify = task.getNotify();
					log.debug("任务被拒绝...重新加入队列: {}, delayMillions={}ms", notify.getLocalOrderSn(), delayMillions);
					redisService.leftpush(NOTIFY_EVENT_QUEUE_NAME, JSON.toJSONString(notify));
				}
			});
		}
		if (queuePopTaskExecutor == null) {
			queuePopTaskExecutor = Executors.newSingleThreadExecutor();
			queuePopTaskfuture = queuePopTaskExecutor.submit(new Runnable() {

				@Override
				public void run() {
					try {
						log.debug("回调车场消息队列启动...");
						while (notifyTaskExitFlag) {
							try {
								int popDelayMillis = popDelayTime.get();
								if (popDelayMillis > 0) {
									Thread.sleep(popDelayMillis);
								}
								// log.debug("redis队列rightpop...");
								// redis最大等待时间为5s
								String data = redisService.rightpop(NOTIFY_EVENT_QUEUE_NAME, 5, TimeUnit.SECONDS);
								if (data != null) {
									log.debug("取到值: {}", data);
									if (data != null) {
										notifyTaskExecutor.submit(new NotifyTask(data));
										if (popDelayTime.get() > 0) {
											popDelayMillis = popDelayTime.getAndDecrement();// 每取一次（不知道成功还是失败）,队列pop间隔时间减少1ms
											log.debug("popDelayTime减速到{}ms", popDelayTime);
										}
									}
								}
							} catch (Exception e) {
								log.error("redis消息队列获取失败:", e);
								try {
									Thread.sleep(5000);
								} catch (InterruptedException e1) {
								}
							}
						}
					} catch (Exception e) {
						log.error("", e);
					}
				}
			});
		}
	}

	class NotifyTask implements Runnable {
		NotifyParkingLotMsg notify;

		public NotifyTask(String data) {
			notify = JSON.parseObject(data, NotifyParkingLotMsg.class);
		}

		@Override
		public void run() {
			PluginService pluginService = SpringUtils.getBean(PluginService.class);
			ParkingPlugin parkingPlugin = pluginService.getParkingPlugin(notify.getPluginId());
			boolean isNotifySuccess = false;
			try {
				if (SettingUtils.get().getIsDebug()) {
					isNotifySuccess = true;
					Thread.sleep(5000);
				} else {
					isNotifySuccess = parkingPlugin.notifyParkingFeeOrderSuccess(notify);
				}
			} catch (Exception e) {
				log.debug("通知车场失败:", e);
				if (e instanceof MyException) {
					notify.notifyFail();
					if (!notify.isNotifyTimeout()) {
						log.debug("未超时，重新加入队列: {}", notify.getLocalOrderSn());
						redisService.leftpush(NOTIFY_EVENT_QUEUE_NAME, JSON.toJSONString(notify));
					} else {
						log.debug("通知超时: {}", JSON.toJSONString(notify));
					}
				}
			}
			if (isNotifySuccess) {
				log.debug("通知成功: {}", notify.getLocalOrderSn());
			}
		}

		public final NotifyParkingLotMsg getNotify() {
			return notify;
		}

		public final void setNotify(NotifyParkingLotMsg notify) {
			this.notify = notify;
		}
	}

	@Override
	public void destroy() throws Exception {
		log.debug("destroy, 关闭ParkingPlugin");
		notifyTaskExitFlag = false;
		try {
			queuePopTaskfuture.cancel(true);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			queuePopTaskExecutor.shutdown();
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			if (notifyTaskExecutor != null)
				notifyTaskExecutor.shutdown();
		} catch (Exception e) {
			log.error("", e);
		}
	}

}
